package cn.com.infcn.spider.scheduler.ftp;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.nlpcn.commons.lang.util.StringUtil;

import cn.com.infcn.spider.SchedulerManager;
import cn.com.infcn.spider.URIManager;
import cn.com.infcn.spider.domain.FtpJob;
import cn.com.infcn.spider.domain.FtpUriInfo;
import cn.com.infcn.spider.io.input.Input;
import cn.com.infcn.spider.io.input.InputFactory;
import cn.com.infcn.spider.io.input.impl.ftpInput;
import cn.com.infcn.spider.io.output.OutputFactory;
import cn.com.infcn.spider.scheduler.Scheduler;
import cn.com.infcn.spider.util.MongoDao;
import cn.com.infcn.spider.util.StaticValue;

import com.alibaba.fastjson.JSONObject;
import com.mongodb.BasicDBObject;
import com.mongodb.DBCollection;

/**
 * job调度类
 * 
 * @author mixl
 *
 */
public class FtpScheduler extends Scheduler {

	private ExecutorService es = null;

	private boolean flag = true;

	private static final int MAX_TIMES = 120;

	public FtpScheduler(FtpJob job) {
		super(job);
	}

	@Override
	public void shutdown() {
		try {
			super.shutdown();
		} catch (Exception e) {
			e.printStackTrace();
			LOG.error(e);
		}
		flag = false;
		es.shutdown();
	}

	@Override
	public void run() {

		JSONObject object = null;
		for (Input input : inputs) {
			if (input instanceof ftpInput) {
				for (Object obj : input.all()) {
					object = (JSONObject)obj;
					String server = object.getString("server");
					int port = object.getIntValue("port");
					String user = object.getString("user");
					String password = object.getString("password");
					String path = object.getString("path");
					URIManager.add(job, new FtpUriInfo(server, port, user, password, path), false);
				}
			}
		}
		
		
		FtpJob job = ((FtpJob) this.job);
		
		int threadNum = job.getThreadNum() ;
		
		// 创建线程队列
		es = Executors.newScheduledThreadPool(threadNum);

		AtomicInteger atomicInteger = new AtomicInteger();
		

		int emptyCount = 0;
		FtpUriInfo uri = null;
		while (flag) {
			
			uri = (FtpUriInfo)URIManager.pop(job);
			
			if(uri == null){
				try {
					Thread.currentThread().sleep(1000);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
				continue;
			}
			
			if (job.getDepth() > 0 && uri.getDepth() > job.getDepth()) {
				LOG.info("job depth is " + job.getDepth() + " url depth is " + uri.getDepth() + " so skip!");
				continue;
			}

			if (StringUtil.isBlank(uri.getUri())) { // 网址为空休息10s
				emptyCount++;
				try {
					Thread.sleep(1000L);
				} catch (InterruptedException e) {
					flag = false;
					break;
				}

				if (emptyCount >= MAX_TIMES) {
					flag = false;
				}

				continue;
			}
			emptyCount = 0;

			atomicInteger.incrementAndGet();
			try {
				
				while (atomicInteger.get() > threadNum) {
					try {
						LOG.debug("run thread:"+atomicInteger.get()+" config threadNumL"+threadNum+" now to sleep 10s!");
						Thread.sleep(10000L);
					} catch (InterruptedException e) {
						flag = false;
						break;
					}
				}

				es.execute(new FtpWorker(this, job, (FtpUriInfo)uri, atomicInteger));

			} catch (Exception e) {
				e.printStackTrace();
			}

		}

		// 说明结束了
		if (emptyCount >= MAX_TIMES) {
			job.setStatus(2);
			StaticValue.getBean(MongoDao.class, "job").save("job", job);
		}

		SchedulerManager.removeKey(job);
	}

	@Override
	public boolean isRuning() {
		return flag;
	}

	@Override
	protected void init() {

		String jobName = job.getName();

		inputs = InputFactory.create(job.getInput(), jobName);

		// 创建MongoDB索引
		MongoDao mongoDao = StaticValue.getBean(MongoDao.class, "mongoDao");
		DBCollection collection = mongoDao.getCollection(URIManager.SUCCESS + jobName);
		if (collection.count() == 0) {
			collection.createIndex(new BasicDBObject("update_time", 1));
		}
		collection = mongoDao.getCollection(URIManager.ERROR + jobName);
		if (collection.count() == 0) {
			collection.createIndex(new BasicDBObject("update_time", 1));
		}

		outputs = OutputFactory.create(job.getOutput(), jobName);
	}

	@Override
	public Object test(String uri) {
		// TODO Auto-generated method stub
		return null;
	}

}
